package com.example.mq.kafka.consumer.ack;

import com.example.mq.kafka.dataobject.Kafka01Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

/**
 * 手动ack
 *
 * @author luowenjie
 * @since 2024/4/9 14:17
 */
@Slf4j
public class KafkaAckConsumer {
    @KafkaListener(topics = Kafka01Message.TOPIC, concurrency = "2",
            groupId = "demo01-consumer-group-" + Kafka01Message.TOPIC)
    public void onMessage(Kafka01Message message, Acknowledgment acknowledgment) {
        try {
            log.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
            // 手动提交 ACK
            acknowledgment.acknowledge();
        } catch (Exception e) {
            log.error("[onMessage][线程编号:{} 消息({})处理失败]", Thread.currentThread().getId(), message, e);
            // 处理失败，可以选择重试或者其他处理方式  kafka 会重复发送该消息
        }
    }
}
